package com.tang.watermake.generator;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

/**
 * 自定义周期性水位线
 * Punctuated /ˈpʌŋktʃueɪtɪd/ 不时打断；给……加标点符号（punctuate 的过去式及过去分词）
 *
 * @author tang
 * @since 2023/6/14 22:35
 */
public class MyPunctuatedWatermarkGenerator<T> implements WatermarkGenerator<T> {

    // 乱序等待时间
    private final Long delayTimestamp;
    // 最大的当前时间
    private Long maxTimestamp;

    public MyPunctuatedWatermarkGenerator(Long delayTimestamp) {
        this.delayTimestamp = delayTimestamp;
        this.maxTimestamp = Long.MIN_VALUE + this.delayTimestamp + 1;
    }

    /**
     * Called for every event, allows the watermark generator to examine and remember the event
     * timestamps, or to emit a watermark based on the event itself.
     *
     * @param event          当前事件对象
     * @param eventTimestamp 当前的时间
     * @param output         Watermark输出对象
     */
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        maxTimestamp = Math.max(eventTimestamp, maxTimestamp);
        output.emitWatermark(new Watermark(maxTimestamp - delayTimestamp - 1));
        System.out.println("调用onEvent方法，获取目前为止的最大时间戳=" + maxTimestamp
                + ",watermark=" + (maxTimestamp - delayTimestamp - 1));
    }

    /**
     * Called periodically, and might emit a new watermark, or not.
     *
     * <p>The interval in which this method is called and Watermarks are generated depends on
     * ExecutionConfig getAutoWatermarkInterval().
     *
     * @param output Watermark输出对象
     */
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
    }

}
